package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/13
 */
public class App04_DwdMhsExamReport extends BaseSqlApp {
    public static void main(String[] args) {
        new App04_DwdMhsExamReport().init(
                5004,
                "App4_DwdMhsExamReport",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target (" +
                "HOSPITAL_NAME                    STRING," +
                "HOSPITAL_CODE                    STRING," +
                "AGE_UNIT                         STRING," +
                "AGE                              STRING," +
                "SEX_NAME                         STRING," +
                "SEX_CODE                         STRING," +
                "ID_CARD_NUMBER                   STRING," +
                "PATIENT_VISIT_CATEGORY_CODE      STRING," +
                "PATIENT_VISIT_CATEGORY           STRING," +
                "VISIT_ID                         STRING," +
                "INPATIENT_VISIT_FLOW_ID          STRING," +
                "INPATIENT_NO                     STRING," +
                "OUTPATIENT_VISIT_FLOW_ID         STRING," +
                "OUTPATIENT_NO                    STRING," +
                "PATIENT_ID                       STRING," +
                "PATIENT_NAME                     STRING," +
                "DOMAIN_ID                        STRING," +
                "VISIT_TIMES                      STRING," +
                "EXAM_REPORT_NO                   STRING," +
                "APPLY_SHEET_NO                   STRING," +
                "SPECIMEN_NUMBER                  STRING," +
                "TELECOM                          STRING," +
                "EXAM_REPORT_TIME                 STRING," +
                "AUTHOR_CODE                      STRING," +
                "AUTHOR_NAME                      STRING," +
                "CUSTODIAN_NAME                   STRING," +
                "CUSTODIAN_CODE                   STRING," +
                "CUSTODIAN_ADDR                   STRING," +
                "AUDIT_SIGNDATE                   STRING," +
                "EXAM_AUDIT_DOCTOR_ID             STRING," +
                "EXAM_AUDIT_DOCTOR_NAME           STRING," +
                "EXAM_AUDIT_DATE                  STRING," +
                "EXAM_TECH_STAFF_ID               STRING," +
                "EXAM_TECH_STAFF_NAME             STRING," +
                "TECHNICIAN_SIGNDATE              STRING," +
                "EXAM_DOCTOR_ID                   STRING," +
                "EXAM_DOCTOR_NAME                 STRING," +
                "APPLY_DATE                       STRING," +
                "APPLY_DEPT                       STRING," +
                "APPLY_AGENCYCODE                 STRING," +
                "APPLY_ORGAN                      STRING," +
                "VISIT_DATE                       STRING," +
                "BED_NO                           STRING," +
                "BED_NAME                         STRING," +
                "ROOM_NO                          STRING," +
                "ROOM_NAME                        STRING," +
                "DEPT_ID                          STRING," +
                "DEPT_NAME                        STRING," +
                "WARDS_CODE                       STRING," +
                "WARDS_NAME                       STRING," +
                "CHIEF_COMPLAINT                  STRING," +
                "SYMPTOM_START_TIME               STRING," +
                "SYMPTOM_END_TIME                 STRING," +
                "SYMPTOM_DESC                     STRING," +
                "EXAM_SPECIAL_FLAG                STRING," +
                "CHECK_METHOD                     STRING," +
                "INSPECT_TYPE                     STRING," +
                "TREAT_PROCESS_DESC               STRING," +
                "AGENCY_NAME                      STRING," +
                "REPORT_REMARK                    STRING," +
                "INSPECT_TYPE_CODE                STRING," +
                "APPLY_DEPT_ID                    STRING," +
                "EXAM_DATE                        STRING," +
                "ITEM_NAME                        STRING," +
                "ITEM_CODE                        STRING," +
                "EXAM_RESULT_VALUE                STRING," +
                "EXAM_RESULT_UNIT                 STRING," +
                "RESULT_CODE                      STRING," +
                "RESULT_NAME                      STRING," +
                "SPECIMEN_STATUS                  STRING," +
                "SPECIMEN_FIXATIVE_NAME           STRING," +
                "SPECIMEN_SAMPLE_TIME             STRING," +
                "SPECIMEN_RECV_TIME               STRING," +
                "SPECIMEN_CLASS                   STRING," +
                "DIAG_CODE                        STRING," +
                "DIAG_NAME                        STRING," +
                "primary key (EXAM_REPORT_NO) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_EXAM_REPORT));
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "visit_id as VISIT_ID," +
                "case when patient_typecode ='3' then visit_id else '' end  as INPATIENT_VISIT_FLOW_ID , " +
                "inpatient as INPATIENT_NO , " +
                "case when patient_typecode <> '3' then visit_id else '' end  as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "visit_times as VISIT_TIMES , " +
                "s.unique_id as EXAM_REPORT_NO , " +
                "appy_id as APPLY_SHEET_NO , " +
                "specimen_number as SPECIMEN_NUMBER , " +
                "telecom as TELECOM , " +
                "write_time as EXAM_REPORT_TIME , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_addr as CUSTODIAN_ADDR , " +
                "audit_signdate as AUDIT_SIGNDATE , " +
                "audit_code as EXAM_AUDIT_DOCTOR_ID , " +
                "audit_name as EXAM_AUDIT_DOCTOR_NAME , " +
                "check_signdate as EXAM_AUDIT_DATE , " +
                "technician_code as EXAM_TECH_STAFF_ID , " +
                "technician_name as EXAM_TECH_STAFF_NAME , " +
                "technician_signdate as TECHNICIAN_SIGNDATE , " +
                "check_code as EXAM_DOCTOR_ID , " +
                "check_name as EXAM_DOCTOR_NAME , " +
                "apply_date as APPLY_DATE , " +
                "apply_dept as APPLY_DEPT , " +
                "apply_agencycode as APPLY_AGENCYCODE , " +
                "apply_organ as APPLY_ORGAN , " +
                "visit_date as VISIT_DATE , " +
                "bed_no as BED_NO , " +
                "bed_name as BED_NAME , " +
                "ward_id as ROOM_NO , " +
                "ward_name as ROOM_NAME , " +
                "dept_code as DEPT_ID , " +
                "dept_name as DEPT_NAME , " +
                "wards_id as WARDS_CODE , " +
                "wards_name as WARDS_NAME , " +
                "chief_complaint as CHIEF_COMPLAINT , " +
                "symptom_start_date as SYMPTOM_START_TIME , " +
                "symptom_stopping_date as SYMPTOM_END_TIME , " +
                "symptom_description as SYMPTOM_DESC , " +
                "cast(case when special_exam_mark='1' then special_exam_mark='true' else special_exam_mark='false' end as string)  as EXAM_SPECIAL_FLAG , " +
                "check_method as CHECK_METHOD , " +
                "inspect_type as INSPECT_TYPE , " +
                "treatment_process as TREAT_PROCESS_DESC , " +
                "agency_name as AGENCY_NAME , " +
                "report_remark as REPORT_REMARK , " +
                "inspect_type_code as INSPECT_TYPE_CODE , " +
                "apply_dept_code as APPLY_DEPT_ID , " +
                "check_date as EXAM_DATE , " +
                "item_name as ITEM_NAME , " +
                "item_code as ITEM_CODE , " +
                "quanty_result as EXAM_RESULT_VALUE , " +
                "quanty_unit as EXAM_RESULT_UNIT , " +
                "result_code as RESULT_CODE , " +
                "result_name as RESULT_NAME , " +
                "specimen_status as SPECIMEN_STATUS , " +
                "specimen_fixative as SPECIMEN_FIXATIVE_NAME , " +
                "sample_date as SPECIMEN_SAMPLE_TIME , " +
                "receipt_date as SPECIMEN_RECV_TIME , " +
                "specimen_category as SPECIMEN_CLASS , " +
                "diag_code as DIAG_CODE , " +
                "diag_name as DIAG_NAME  " +
                "from hdsd00_05_02 s    " +
                "left join tdy_list_exam e on s.unique_id = e.unique_id and s.xds_type = e.xds_type    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    ");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_05_02
        tableEnv.executeSql("create table hdsd00_05_02 (" +
                "pk                         string," +
                "upload_time                string," +
                "status                     string," +
                "empi                       string," +
                "encounter_id               string," +
                "visit_domain               string," +
                "visit_id                   string," +
                "visit_times                string," +
                "patient_id                 string," +
                "patient_domain             string," +
                "unique_id                  string," +
                "xds_type                   string," +
                "xds_name                   string," +
                "effective_time             string," +
                "xds_version                string," +
                "domain_id                  string," +
                "out_patient_id             string," +
                "inpatient                  string," +
                "patient_type               string," +
                "report_id                  string," +
                "patient_typecode           string," +
                "appy_id                    string," +
                "specimen_number            string," +
                "telecom                    string," +
                "identify_no                string," +
                "patient_name               string," +
                "sex                        string," +
                "sex_code                   string," +
                "age                        string," +
                "age_unit                   string," +
                "write_time                 string," +
                "author_code                string," +
                "author_name                string," +
                "custodian_name             string," +
                "custodian_code             string," +
                "custodian_addr             string," +
                "audit_signdate             string," +
                "audit_code                 string," +
                "audit_name                 string," +
                "check_signdate             string," +
                "technician_code            string," +
                "technician_name            string," +
                "technician_signdate        string," +
                "check_code                 string," +
                "check_name                 string," +
                "apply_date                 string," +
                "apply_dept_code            string," +
                "apply_dept                 string," +
                "apply_agencycode           string," +
                "apply_organ                string," +
                "visit_date                 string," +
                "bed_no                     string," +
                "ward_id                    string," +
                "ward_name                  string," +
                "wards_id                   string," +
                "bed_name                   string," +
                "dept_code                  string," +
                "dept_name                  string," +
                "wards_name                 string," +
                "hospital_code              string," +
                "hospital_name              string," +
                "chief_complaint            string," +
                "symptom_start_date         string," +
                "symptom_stopping_date      string," +
                "symptom_description        string," +
                "special_exam_mark          string," +
                "check_method               string," +
                "inspect_type               string," +
                "treatment_process          string," +
                "objective_observation      string," +
                "subjective_tip             string," +
                "report_dept                string," +
                "agency_name                string," +
                "report_remark              string," +
                "inspect_type_code          string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_05_02","App4_DwdMhsExamReport"));
        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App4_DwdMhsExamReport");
        // tdy_list_exam
        super.readTdyListExam(tableEnv,"App4_DwdMhsExamReport");
    }
}
